package top.jolyoulu.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Author: JolyouLu
 * @Date: 2024/5/16 19:21
 * @Description
 */
object Spark01_SparkSQL_Transform {
  def main(args: Array[String]): Unit = {
    //创建SparkSQL的运行环境
    val sparkSQL: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")
    val spark = SparkSession.builder().config(sparkSQL).getOrCreate()
    val sc: SparkContext = spark.sparkContext
    import spark.implicits._ //将spark转换规则引入
    //DataFrame转换
    val path: String = this.getClass.getClassLoader.getResource("datas/user.json").toURI.getPath
    val df: DataFrame = spark.read.json(path)
    println("======================RDD => DataFrame======================")
    val rdd2df: DataFrame = sc.makeRDD(List(("张三", 30), ("李四", 25), ("王五", 30))).toDF()
    rdd2df.show()
    println("======================DataFrame => RDD======================")
    val df2rdd: RDD[Row] = df.rdd
    df2rdd.foreach(println)
    println("======================DataFrame => DataSet======================")
    val df2ds: Dataset[User] = df.as[User]
    df2ds.show()

    //DataSet转换
    val ds: Dataset[User] = df.as[User]
    println("======================RDD => DataSet======================")
    val rdd2ds: Dataset[User] = sc.makeRDD(List(("张三", 30), ("李四", 25), ("王五", 30)))
      .map(item => User(item._2, item._1)).toDS()
    rdd2ds.show()
    println("======================DataSet => RDD======================")
    val ds2rdd: RDD[User] = ds.rdd
    ds2rdd.foreach(println)
    println("======================DataSet => DataFrame======================")
    val ds2df: DataFrame = ds.toDF()
    ds2df.show()
    
    //关闭环境
    spark.close()
  }
  case class User(age:Long,username:String)
}
